package org.lib.hadoop.hello;

import java.io.IOException;
import java.net.URI;
import java.util.Date;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import lombok.extern.slf4j.Slf4j;

/**
 * Hadoop运行器
 * @author ben
 * @date 2021-12-07 20:17:41 CST
 */
@Component
@Slf4j
public class HadoopRunner implements ApplicationRunner {

	private static String haddr = "hdfs://localhost:19870";
	
	@Override
	public void run(ApplicationArguments args) throws Exception {
		Configuration hconf = new Configuration();
		
		FileSystem fs = FileSystem.get(new URI(haddr), hconf);
		System.out.println("fs=" + fs);
		
		String newDir = "/sbdir";
		makeDir(fs, newDir);
		makeDir(fs, "/y12d09");
		
		// 拷贝文件
		final String fn = "text1.txt";
		testCopyFromLocalFile(fs, AppConstants.HSRC + fn, newDir);
		
		String hfile = newDir + "/" + fn;
		System.out.println("hfile=" + hfile);
		copyToLocal(fs, hfile, "/home/hadoop/hdst");
		fileStatus(fs, hfile);
		
		moveToLocal(fs, hfile, "/home/hadoop/hdst");
		fileStatus(fs, hfile);
				
		listFiles(fs, "/");
		listFiles(fs, "/user");
		listFiles(fs, newDir);
		// java.io.FileNotFoundException
//		listFiles(fs, "/notExists");
		
		fs.printStatistics();
		
		makeFileAndRw(fs, "/file1", "myfile is a hdfs file.");
		makeFileAndRw(fs, newDir + "/file2", "我的文件，使用HDFS创建的文件");
		
		// 1222
		this.delFile(fs, "/file1");
		this.delFile(fs, "/file1");
		this.delDirectory(fs, newDir);
		this.delDirectory(fs, newDir);
		
		fs.close();
	}
	
	/**
	 * 创建文件夹
	 * @author ben
	 * @date 2021-12-09 20:16:59 CST
	 * @param fs
	 * @param dirPath
	 */
	private void makeDir(FileSystem fs, String dirPath) {
		if (StringUtils.isBlank(dirPath)) {
			return;
		}
		if (!dirPath.startsWith("/")) {
			dirPath = "/" + dirPath;
		}
		try {
			if (!fs.exists(new Path(dirPath))) {
				boolean mkdirsRet = fs.mkdirs(new Path(dirPath));
				System.out.println("mkdirsRet=" + mkdirsRet);
				
				if (mkdirsRet) {
					System.out.println("OK：创建文件夹成功！");
				} else {
					System.out.println("ERROR：创建文件夹失败！");
				}
			} else {
				System.out.println("文件夹已存在：dirPath=" + dirPath);
			}
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * listFiles
	 * @author ben
	 * @date 2021-12-09 20:23:27 CST
	 * @param fs
	 * @param dirUser
	 */
	private void listFiles(FileSystem fs, String dirUser) {
		System.out.println("listFiles: dirUser=" + dirUser);
		try {
//			RemoteIterator<LocatedFileStatus> itor = fs.listFiles(new Path(dirUser), false);
			RemoteIterator<LocatedFileStatus> itor = fs.listFiles(new Path(dirUser), true);
			while (itor.hasNext()) {
				LocatedFileStatus lfs = itor.next();
				System.out.println("\nlfs=" + lfs);
				log.info("lfs: {}, {}, {}, {}, {}, {}", lfs.getPath().getName(),
						lfs.getBlockSize(),
						lfs.getLen(),
						lfs.getGroup(),
						lfs.getOwner(),
						lfs.getReplication());
			}
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
		System.out.println("listFiles: END");
	}
	
	/**
	 * 拷贝本地文件到HDFS？
	 * @author ben
	 * @date 2021-12-09 20:24:38 CST
	 * @param fs
	 * @param path1
	 * @param path2
	 */
	private void testCopyFromLocalFile(FileSystem fs, String path1, String path2) {
		try {
			fs.copyFromLocalFile(new Path(path1), new Path(path2));
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * 拷贝HDFS文件到本地
	 * @author ben
	 * @date 2021-12-11 20:42:44 CST
	 * @param fs
	 * @param file
	 * @param localPath
	 */
	private void copyToLocal(FileSystem fs, String file, String localPath) {
		try {
			fs.copyToLocalFile(new Path(file), new Path(localPath));
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * 移动HDFS文件到本地
	 * @author ben
	 * @date 2021-12-11 20:44:49 CST
	 * @param fs
	 * @param file
	 * @param localPath
	 */
	private void moveToLocal(FileSystem fs, String file, String localPath) {
		try {
			fs.moveToLocalFile(new Path(file), new Path(localPath));
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * 文件状态检查
	 * @author ben
	 * @date 2021-12-11 20:45:40 CST
	 * @param fs
	 * @param file
	 */
	private void fileStatus(FileSystem fs, String file) {
		try {
			boolean ret = fs.exists(new Path(file));
			System.out.println(file + "-exists: " + ret);
			
			if (ret) {
				FileStatus status = fs.getFileStatus(new Path(file));
				System.out.println("status=" + status);
				System.out.println("status.owner=" + status.getOwner());
				System.out.println("status.len=" + status.getLen());
			}
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * 创建文件&读写
	 * @author ben
	 * @date 2021-12-13 20:25:35 CST
	 * @param fs HDFS
	 * @param file 文件路径
	 * @param content 文件内容
	 */
	private void makeFileAndRw(FileSystem fs, String file, String content) {
		System.out.println("\nmakeFileAndRw: file=" + file);
		
		Path fpath = new Path(file);
		try {
			boolean exists = fs.exists(fpath);
			if (exists) {
				System.out.println("文件已存在，开始读取...");
				
				FSDataInputStream fsdis = fs.open(fpath);
				int bytes = fsdis.available();
				System.out.println("bytes=" + bytes);
				
				// read
				int bt = 0;
				while ((bt = fsdis.read()) != -1) {
					System.out.printf("bt=0x%02x, post=%d\r\n", bt, fsdis.getPos());
				}
				
				fsdis.close();
				System.out.println("文件已读完，关闭。");
			} else {
				System.out.println("文件不存在：开始创建");
				FSDataOutputStream fsdos = fs.create(fpath);
				System.out.println("fsdos=" + fsdos);
				
				fsdos.writeChars(content);
				System.out.println("文件写入成功！");
				fsdos.writeUTF(content);
				System.out.println("文件写入成功X2！");
				fsdos.writeLong(System.currentTimeMillis());
				System.out.println("文件写入成功X3！");
				fsdos.writeUTF(new Date().toGMTString());
				System.out.println("文件写入成功X4！");
				
				fsdos.flush();
				System.out.println("flushed!");
				fsdos.close();
				System.out.println("closed!");
			}
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * 删除文件
	 * @author ben
	 * @date 2021-12-22 20:27:59 CST
	 * @param fs
	 * @param file
	 */
	private void delFile(FileSystem fs, String file) {
		log.info("delFile: file={}", file);
		try {
			Path fp = new Path(file);
			if (fs.exists(fp)) {
				// 删除文件，使用false
				boolean delRet = fs.delete(fp, false);
				log.info("delFile：删除文件结果，delRet={}", delRet);
			} else {
				log.info("delFile：文件不存在");
			}
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}
	
	/**
	 * 删除文件夹
	 * @author ben
	 * @date 2021-12-22 20:28:07 CST
	 * @param fs
	 * @param dir
	 */
	private void delDirectory(FileSystem fs, String dir) {
		log.info("delDirectory: dir={}", dir);
		try {
			Path fp = new Path(dir);
			if (fs.exists(fp)) {
				// 循环删除文件夹下的内容
				boolean delRet = fs.delete(fp, true);
				log.info("delDirectory：删除目录结果，delRet={}", delRet);
			} else {
				log.info("delDirectory：目录不存在");
			}
		} catch (IllegalArgumentException | IOException e) {
			log.error("发生异常：e={}", e);
		}
	}

}
